package com.qupeng.demo.kafka.kafkaspringbootconsumer.model;

import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.Data;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.record.TimestampType;

import java.io.Serializable;
import java.util.*;

@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class ConsumeSuccessResponse<K, V> implements Serializable {

    private List<ConsumerRecordData<K, V>> consumerRecordDatas = new ArrayList<>();

    public ConsumeSuccessResponse() {

    }

    public ConsumeSuccessResponse(ConsumerRecords<K, V> consumerRecords) {
        consumerRecords.forEach(consumerRecord -> consumerRecordDatas.add(new ConsumerRecordData<>(consumerRecord)));
    }

    @Data
    private static class ConsumerRecordData<K, V> implements Serializable {
        public ConsumerRecordData(ConsumerRecord<K, V> consumerRecord) {
            this.topic = consumerRecord.topic();
            this.partition = consumerRecord.partition();
            Optional.ofNullable(consumerRecord.headers()).ifPresentOrElse(headers -> {
                headers.forEach(header -> this.headers.put(header.key(), new String(header.value())));
            }, () -> {});
            this.key = consumerRecord.key();
            this.value = consumerRecord.value().toString();
            this.timestamp = new Date(consumerRecord.timestamp());
            this.timestampType = consumerRecord.timestampType();
            this.leaderEpoch = consumerRecord.leaderEpoch().map(String::valueOf).orElse("null");
        }

        String topic;
        int partition;
        Map<String, String> headers = new HashMap<>();
        K key;
        String value;
        @JsonFormat(timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
        Date timestamp;
        TimestampType timestampType;
        String leaderEpoch;
    }
}
